package com.hjs.mr.merge_and_sort;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

/**
 * 整个MapReduce的驱动类，定义了再执行MapReduce任务时需要用到的所有相关类和数据的来源和输出位置
 * 整个MpaReduce的任务过程是首先将三个文件中的内容进行统一的读取，将读取的结果定义成一个Reduce任务，这是通过分区器将三个文件的读取结果分到一个分区实现的
 * 因为Map阶段读取的数据被封装到了SortBean中，并且SortBean实现了WritableComparable接口，实现了排序的方法，所以在数据读取完成之后就已经有顺序了
 * 之后Reduce中因为定义了OutputFormat，所以会按照自定义的OutputFormat进行输出，并且因为分区器将所有数据输出到了一个Reduce任务重，所以可以在Reduce阶段为SortBean中的排序字段进行赋值
 * 完成之后直接将SortBean输出即可，因为在SortBean中自定义了toString方法，所以会按照该格式输出
 */
public class SortDriver {

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration, "SortDriver");

        job.setJarByClass(SortDriver.class);

        job.setMapperClass(SortMapper.class);
        job.setReducerClass(SortReducer.class);

        job.setMapOutputKeyClass(SortBean.class);
        job.setMapOutputValueClass(NullWritable.class);

        job.setOutputKeyClass(SortBean.class);
        job.setOutputValueClass(NullWritable.class);

        //指定使用自定义的分区器
        job.setPartitionerClass(SortPartitioner.class);

        //指定使用自定义的InputFormat和OutputFormat
        job.setInputFormatClass(MergeInputFormat.class);
        job.setOutputFormatClass(SortOutputFormat.class);

        //指定数据来源以及结果的输出位置
        FileInputFormat.setInputPaths(job,new Path("D:/HomeworkData"));
        FileOutputFormat.setOutputPath(job,new Path("D:/HomeworkResult/sortOutput"));

        boolean flag = job.waitForCompletion(true);
        System.exit(flag?0:1);
    }
}
